﻿namespace Mueller.CoffeeMonitor
#nowarn "40"

/// Module for working with Reactive Extensions (IObservable)
module Observable =

    open System
    open System.Reactive.Linq

    /// Buffers the observable for the specified interval
    let inline buffer (interval:TimeSpan) observable =
        Observable.Buffer(observable, interval)

    /// Merges a set of observables into a single observable.
    let inline mergeAll (observables: seq<IObservable<_>>) =
        Observable.Merge(observables)

    let inline amb (first: IObservable<_>) (second: IObservable<_>) =
        Observable.Amb(first, second)


[<AutoOpen>]
module Utils =
    open System

    let inline (&=) x y = Object.ReferenceEquals(x, y)

    let inline isNull x = x &= null

    let asOpt (s:string) =
        if String.IsNullOrWhiteSpace s then None
        else Some s

    let (|?) = defaultArg

    type Microsoft.FSharp.Control.Async with 

        static member AwaitObservable(ev1:IObservable<'a>) = 
            Async.FromContinuations((fun (cont,econt,ccont) -> 
                let rec callback = (fun value ->
                    remover.Dispose()
                    cont(value) )
                and remover : IDisposable  = ev1.Subscribe(callback) 
                () ))
  
        static member AwaitObservable(ev1:IObservable<'a>, ev2:IObservable<'b>) = 
            Async.FromContinuations((fun (cont,econt,ccont) -> 
                let rec callback1 = (fun value ->
                    remover1.Dispose()
                    remover2.Dispose()
                    cont(Choice1Of2(value)) )
                and callback2 = (fun value ->
                    remover1.Dispose()
                    remover2.Dispose()
                    cont(Choice2Of2(value)) )
                and remover1 : IDisposable  = ev1.Subscribe(callback1) 
                and remover2 : IDisposable  = ev2.Subscribe(callback2) 
                () ))

        static member AwaitObservable(ev1:IObservable<'a>, ev2:IObservable<'b>, ev3:IObservable<'c>) = 
            Async.FromContinuations((fun (cont,econt,ccont) -> 
                let rec callback1 = (fun value ->
                    remover1.Dispose()
                    remover2.Dispose()
                    remover3.Dispose()
                    cont(Choice1Of3(value)) )
                and callback2 = (fun value ->
                    remover1.Dispose()
                    remover2.Dispose()
                    remover3.Dispose()
                    cont(Choice2Of3(value)) )
                and callback3 = (fun value ->
                    remover1.Dispose()
                    remover2.Dispose()
                    remover3.Dispose()
                    cont(Choice3Of3(value)) )
                and remover1 : IDisposable  = ev1.Subscribe(callback1) 
                and remover2 : IDisposable  = ev2.Subscribe(callback2) 
                and remover3 : IDisposable  = ev3.Subscribe(callback3) 
                () ))

